package com.hyk.learning.proxy;

import com.hyk.learning.circuitBreaker.CircuitBreaker;
import com.hyk.learning.common.enums.LoadBalanceEnum;
import com.hyk.learning.common.enums.RpcErrorMessageEnum;
import com.hyk.learning.common.enums.RpcResponseCodeEnum;
import com.hyk.learning.common.enums.ServiceDiscoveryEnum;
import com.hyk.learning.common.exception.RpcException;
import com.hyk.learning.common.extension.ExtensionLoader;
import com.hyk.learning.common.factory.SingletonFactory;
import com.hyk.learning.common.utils.AdaptiveMetrics;
import com.hyk.learning.config.RpcServiceConfig;
import com.hyk.learning.dto.RpcRequest;
import com.hyk.learning.dto.RpcResponse;
import com.hyk.learning.provider.CircuitBreakerProvider;
import com.hyk.learning.registry.ServiceDiscovery;
import com.hyk.learning.remoting.retry.guavaRetry;
import com.hyk.learning.remoting.transport.RpcRequestTransport;
import com.hyk.learning.remoting.transport.netty.client.NettyRpcClient;
import com.hyk.learning.remoting.transport.socket.SocketRpcClient;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

/**
 * @className: RpcClientProxy
 * @author: hyk199710
 * @description: Dynamic proxy class. When a dynamic proxy object calls a method, it actually calls the following invoke method.
 * It is precisely because of the dynamic proxy that the remote method called by the client is like calling the local method (the intermediate process is shielded)
 * @date: 2024/5/19 18:07
 */
@Slf4j
public class RpcClientProxy implements InvocationHandler {

    private static final String INTERFACE_NAME = "interfaceName";

    /**
     * Used to send requests to the server.And there are two implementations: socket and netty
     */
    private final RpcRequestTransport rpcRequestTransport;

    private final RpcServiceConfig rpcServiceConfig;

    private AdaptiveMetrics adaptiveMetrics;

    private ServiceDiscovery serviceDiscovery;

    private String loadBalance;

    private CircuitBreakerProvider circuitBreakerProvider;

    public RpcClientProxy(RpcRequestTransport rpcRequestTransport, RpcServiceConfig rpcServiceConfig) {
        this.rpcRequestTransport = rpcRequestTransport;
        this.rpcServiceConfig = rpcServiceConfig;
        this.adaptiveMetrics = SingletonFactory.getInstance(AdaptiveMetrics.class);
        this.circuitBreakerProvider = SingletonFactory.getInstance(CircuitBreakerProvider.class);
        this.serviceDiscovery = ExtensionLoader.getExtensionLoader(ServiceDiscovery.class)
                .getExtension(ServiceDiscoveryEnum.ZK.getName());
        this.loadBalance = LoadBalanceEnum.LOADBALANCE.getName();
    }

    public RpcClientProxy(RpcRequestTransport rpcRequestTransport) {
        this.rpcRequestTransport = rpcRequestTransport;
        this.rpcServiceConfig = new RpcServiceConfig();
    }

    /**
     * @param proxy
     * @param method
     * @param args
     * @return Object
     * @author 10091
     * @description This method is actually called when you use a proxy object to call a method.
     * The proxy object is the object you get through the getProxy method.
     * @date 2024/5/19 18:13
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        log.info("invoked method: [{}]", method.getName());
        long startTime = System.currentTimeMillis();
        RpcRequest rpcRequest = RpcRequest.builder()
                .methodName(method.getName())
                .parameters(args)
                .interfaceName(method.getDeclaringClass().getName())
                .paramTypes(method.getParameterTypes())
                .requestId(UUID.randomUUID().toString())
                .group(rpcServiceConfig.getGroup())
                .version(rpcServiceConfig.getVersion())
                .weight(rpcServiceConfig.getWeight())
                .warmupTS(rpcServiceConfig.getWarmupTS())
                .timeout(rpcServiceConfig.getTimeout())
                .startTime(startTime).build();
        CircuitBreaker circuitBreaker = this.circuitBreakerProvider.getCircuitBreaker(rpcRequest.getRpcServiceName());
        if(!circuitBreaker.allowRequest()){
            return null;
        }
        RpcResponse<Object> rpcResponse = null;
        if (rpcRequestTransport instanceof NettyRpcClient) {
            CompletableFuture<RpcResponse<Object>> completableFuture = null;
            if(this.serviceDiscovery.checkRetry(rpcRequest.getRpcServiceName())){
                completableFuture = (CompletableFuture<RpcResponse<Object>>) new guavaRetry().sendServiceWithRetry(rpcRequest, rpcRequestTransport);
            }
            else{
                completableFuture = (CompletableFuture<RpcResponse<Object>>) rpcRequestTransport.sendRpcRequest(rpcRequest);
            }
            rpcResponse = completableFuture.get();
        }
        if (rpcRequestTransport instanceof SocketRpcClient) {
            rpcResponse = (RpcResponse<Object>) rpcRequestTransport.sendRpcRequest(rpcRequest);
        }
        this.check(rpcResponse, rpcRequest);
        return rpcResponse.getData();
    }

    private void check(RpcResponse<Object> rpcResponse, RpcRequest rpcRequest) {
        if (null == rpcResponse) {
            if("adaptive".equals(this.loadBalance)){
                this.adaptiveMetrics.addErrorReq(rpcRequest.getRpcServiceName());
            }
            throw new RpcException(RpcErrorMessageEnum.SERVICE_INVOCATION_FAILURE, INTERFACE_NAME + ":" + rpcRequest.getInterfaceName());
        }

        if (!rpcRequest.getRequestId().equals(rpcResponse.getRequestId())) {
            if("adaptive".equals(this.loadBalance)){
                this.adaptiveMetrics.addErrorReq(rpcRequest.getRpcServiceName());
            }
            throw new RpcException(RpcErrorMessageEnum.REQUEST_NOT_MATCH_RESPONSE, INTERFACE_NAME + ":" + rpcRequest.getInterfaceName());
        }

        if (null == rpcResponse.getCode() || !rpcResponse.getCode().equals(RpcResponseCodeEnum.SUCCESS.getCode())) {
            if("adaptive".equals(this.loadBalance)){
                this.adaptiveMetrics.addErrorReq(rpcRequest.getRpcServiceName());
            }
            throw new RpcException(RpcErrorMessageEnum.SERVICE_INVOCATION_FAILURE, INTERFACE_NAME + ":" + rpcRequest.getInterfaceName());
        }

        //自适应负载均衡算法ewma参数计算
        if(rpcResponse.getCode().equals(RpcResponseCodeEnum.SUCCESS.getCode()) && "adaptive".equals(this.loadBalance)){
            Map<String, String> metricsMap = rpcResponse.getMatricMap();
            long startTime = rpcRequest.getStartTime();
            if(startTime != 0){
                metricsMap.put("rt", String.valueOf(System.currentTimeMillis() - startTime));
            }
            this.adaptiveMetrics.addConsumerSuccess(rpcRequest.getRpcServiceName());
            this.adaptiveMetrics.setProviderMetrics(rpcRequest.getRpcServiceName(), metricsMap);
        }
    }

    /**
     * @param clazz:
     * @return T
     * @author 10091
     * @description get the proxy object
     * @date 2024/5/19 18:12
     */
    public <T> T getProxy(Class<T> clazz) {
        return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[]{clazz}, this);
    }
}
